package com.zbkj.admin.disruptor;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.annotation.Configuration;

import java.util.Arrays;
import java.util.Objects;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Slf4j
@Getter
@Configuration
public class DisruptorConfig implements InitializingBean, DisposableBean {

    private Disruptor<OrderEvent> disruptor;
    // 添加监控线程
    private final ScheduledExecutorService monitorExecutor = Executors.newSingleThreadScheduledExecutor();
    private static final ThreadPoolExecutor executor2 = new ThreadPoolExecutorFactory(
            ThreadPoolExecutorBuilder.builder()
                    .blockingQueueTypeEnum(BlockingQueueTypeEnum.ARRAY_BLOCKING_QUEUE)
                    .defaultCoreSize(300)
                    .keepAliveTime(60)
                    .unit(TimeUnit.SECONDS)
                    .maxQueueSize(1500)
                    .rejectedPolicyTypeEnum(RejectedPolicyTypeEnum.SYNC_PUT_QUEUE_POLICY)
                    .threadFactory(new CustomThreadFactory(CustomThreadBuilder.builder()
                            .name("order-disruptor-").isDaemon(Boolean.TRUE).build()))
                    .build()).createThreadPoolExecutor();


    /**
     * Invoked by the containing {@code BeanFactory} on destruction of a bean.
     *
     * @throws Exception in case of shutdown errors. Exceptions will get logged
     *                   but not rethrown to allow other beans to release their resources as well.
     */
    @Override
    public void destroy() throws Exception {
        log.debug("[Disruptor关闭]");
        if (disruptor != null) {
            try {
                RingBuffer<OrderEvent> ringBuffer = disruptor.getRingBuffer();
                log.debug("[Disruptor关闭] 剩余未处理事件:{}",
                        ringBuffer.getBufferSize() - ringBuffer.remainingCapacity());

                disruptor.shutdown(5, TimeUnit.SECONDS);
            } catch (Exception e) {
                log.warn("[Disruptor强制关闭] 未处理事件:{}",
                        disruptor.getRingBuffer().remainingCapacity());
                disruptor.halt();
            }
        }
        if (monitorExecutor != null) {
            monitorExecutor.shutdownNow();
        }
        executor2.shutdownNow();
    }

    /**
     * Invoked by the containing {@code BeanFactory} after it has set all bean properties
     * and satisfied {@link BeanFactoryAware}, {@code ApplicationContextAware} etc.
     * <p>This method allows the bean instance to perform validation of its overall
     * configuration and final initialization when all bean properties have been set.
     *
     * @throws Exception in the event of misconfiguration (such as failure to set an
     *                   essential property) or if initialization fails for any other reason
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        int bufferSize = 2048;
        disruptor = new Disruptor<>(OrderEvent::new, bufferSize,
//                Thread.ofVirtual().name("order-disruptor-", 0).factory(),
                executor2.getThreadFactory(),
                ProducerType.MULTI, new BlockingWaitStrategy());
        disruptor.handleEventsWith(new OrderEventHandler());
        disruptor.setDefaultExceptionHandler(new OrderExceptionHandler());
        disruptor.start();
        monitorExecutor.scheduleAtFixedRate(() -> {
            RingBuffer<OrderEvent> ringBuffer = disruptor.getRingBuffer();
            long remaining = ringBuffer.remainingCapacity();

            log.warn("[Disruptor内存预警] 剩余容量仅剩 {}（总容量：{}）", remaining, bufferSize);
        }, 5, 5, TimeUnit.SECONDS); // 每5秒检查一次
//        final int bufferSize = 4096; // 调整为2的幂次方，提升位运算性能
//        final int parallelism = Runtime.getRuntime().availableProcessors() * 2; // 虚拟线程数=CPU核心数*2
//
//        disruptor = new Disruptor<>(
//                OrderEvent::new,
//                bufferSize,
//                Thread.ofVirtual().name("order-disruptor-", 0).factory(),
//                ProducerType.MULTI,
//                new TimeoutBlockingWaitStrategy(100, TimeUnit.MILLISECONDS) // 带超时的等待策略
//        );
//
//        // 创建并行消费者池（需确保OrderEventHandler线程安全）
//        OrderEventHandler[] handlers = new OrderEventHandler[parallelism];
//        Arrays.setAll(handlers, i -> new OrderEventHandler());
//
//        disruptor.handleEventsWith(handlers); // 工作池模式
//        disruptor.setDefaultExceptionHandler(new OrderExceptionHandler());
//        // 5. 配置生产者的背压控制
//        disruptor.start();
/*        // 添加溢出监控
        disruptor.getRingBuffer().addGatingSequences(new Sequence() {
            @Override
            public long get() {
                long remaining = disruptor.getRingBuffer().remainingCapacity();
                if (remaining < bufferSize * 0.1) { // 剩余容量不足10%时告警
                    log.warn("[Disruptor告警] 缓冲区即将满载 剩余容量:{}", remaining);
                }
                return super.get();
            }
        });*/

//        disruptor.start();
//        log.info("订单Disruptor初始化完成 缓冲区:{} 并行度:{}", bufferSize, parallelism);
    }

    // 异常处理器（需新增类）
    @Slf4j
    public static class OrderExceptionHandler implements ExceptionHandler<OrderEvent> {
        @Override
        public void handleEventException(Throwable ex, long sequence, OrderEvent event) {
            log.error("[订单事件异常] seq:{} orderNo:{}", sequence, event.getId(), ex);
        }

        @Override
        public void handleOnStartException(Throwable ex) {
            log.error("[Disruptor启动异常]", ex);
        }

        @Override
        public void handleOnShutdownException(Throwable ex) {
            log.error("[Disruptor关闭异常]", ex);
        }
    }
}
